package momo.app.comms.monitor;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import momo.app.multicast.Msg;
import momo.app.multicast.MsgRcvListener;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;


public class TaskMonitor {
	
	private Log log = LogFactory.getLog(TaskMonitor.class);
	private Log logMonitor = LogFactory.getLog("MoMoTaskMonitorLog");
	
	private ExecutorService executor;
	private LinkedBlockingQueue<Future<?>> queue;
	private boolean statusMonitorInterrupted = false;
	private Thread statusMonitor;
	
	public TaskMonitor() {
		queue = new LinkedBlockingQueue<Future<?>>();
		executor = Executors.newSingleThreadExecutor();
		setStatusMonitor();
	}
	
	public void putTask(Runnable task) {
		Future<?> result = executor.submit(task);
		try {
			queue.put(result);
			
			if ( log.isDebugEnabled() )
				log.debug("Task " + result.hashCode() + " in queue");
			if ( logMonitor.isDebugEnabled() )
				logMonitor.debug("Task " + result.hashCode() + " in queue");
		} catch (InterruptedException e) {
			if ( log.isErrorEnabled() )
				log.error("InterruptedException", e);
			if ( logMonitor.isErrorEnabled() )
				logMonitor.error("InterruptedException", e);
		}
	}
	
	public void putTask(MsgRcvListener listener, Msg msg) {
		Thread task = new MsgListenerTask(listener, msg);
		
		Future<?> result = executor.submit(task);
		try {
			queue.put(result);
			
			if ( log.isDebugEnabled() )
				log.debug("Task " + result.hashCode() + ", MsgId " + msg.getId() + " in queue");
			if ( logMonitor.isDebugEnabled() )
				logMonitor.debug("Task " + result.hashCode() + ", MsgId " + msg.getId() + " in queue");
		} catch (InterruptedException e) {
			if ( log.isDebugEnabled() )
				log.error("InterruptedException", e);
			if ( logMonitor.isDebugEnabled() )
				logMonitor.error("InterruptedException", e);
		}
	}
	
	private void setStatusMonitor() {
		statusMonitor = new Thread() {
			public void run() {
				
				if ( log.isDebugEnabled() )
					log.debug("Status Monitoring starts");
				if ( logMonitor.isDebugEnabled() )
					logMonitor.debug("Status Monitoring starts");
				
				while ( !statusMonitorInterrupted ) {
					try {
						Future<?> result = queue.poll(2, TimeUnit.SECONDS);
						if ( result != null ) {
							
							if ( log.isDebugEnabled() )
								log.debug("Task " + result.hashCode() + " starts monitoring");
							if ( logMonitor.isDebugEnabled() )
								logMonitor.debug("Task " + result.hashCode() + " starts monitoring");
							
							monitorStatus(result);
						}
					} catch (InterruptedException e) {
						statusMonitorInterrupted = true;
						
						if ( log.isErrorEnabled() )
							log.error("InterruptedException", e);
						if ( logMonitor.isErrorEnabled() )
							logMonitor.error("InterruptedException", e);
					}
				}
				if ( log.isDebugEnabled() )
					log.debug("Status Monitoring Ends");
				if ( logMonitor.isDebugEnabled() )
					logMonitor.debug("Status Monitoring Ends");
			}
		};
		statusMonitor.start();
	}
	
	private void monitorStatus(Future<?> result) {
		
		if ( log.isDebugEnabled() )
			log.debug("Monitoring Task " + result.hashCode());
		if ( logMonitor.isDebugEnabled() )
			logMonitor.debug("Monitoring Task " + result.hashCode());
		
		boolean hasTaskDone = false;
		while ( !hasTaskDone && !statusMonitorInterrupted ) {
			try {
				result.get(5, TimeUnit.SECONDS);
				hasTaskDone = true;
				
				if ( log.isDebugEnabled() )
					log.debug("Task " + result.hashCode() + " ends");
				if ( logMonitor.isDebugEnabled() )
					logMonitor.debug("Task " + result.hashCode() + " ends");
				
			} catch (InterruptedException e) {
				hasTaskDone = true;
				
				if ( log.isErrorEnabled() )
					log.error("InterruptedException", e);
				if ( logMonitor.isErrorEnabled() )
					logMonitor.error("InterruptedException", e);
				
			} catch (ExecutionException e) {
				hasTaskDone = true;
				
				if ( log.isErrorEnabled() )
					log.error("ExecutionException", e);
				if ( logMonitor.isErrorEnabled() )
					logMonitor.error("ExecutionException", e);
				
			} catch (TimeoutException e) {
				if ( log.isDebugEnabled() )
					log.debug("Task " + result.hashCode() + " is still running");
				if ( logMonitor.isDebugEnabled() )
					logMonitor.debug("Task " + result.hashCode() + " is still running");
			}
		}
	}
	
	private class MsgListenerTask extends Thread
	{
		private MsgRcvListener listener;
		private Msg msg;
		
		public MsgListenerTask(MsgRcvListener listener, Msg msg)
		{
			this.listener = listener;
			this.msg = msg;
		}
		
		public void run()
		{
			listener.msgReceived(msg);
		}
	}
}//end of class TaskMonitor
